package job.gfp.ranks.get_ready;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.OutputStreamWriter;

import mapred.SimpleJob;
import mapred.config.Compress;
import mapred.config.JobConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.RunningJob;

@SuppressWarnings("unchecked")
public class GetReady extends SimpleJob {

    public GetReady(String... settings) throws IOException {
        super("Get ready", settings);

        this.forceUserSetting("-reduce", 0);
    }

    @Override
    protected Class<? extends Reducer> getCombinerClass() {
        return null;
    }

    @Override
    protected Compress getCompress() {
        return Compress.GZIP;
    }

    @Override
    protected String[] getInputPaths(FileSystem fs) throws IOException {
        return new String[] { this.getUserSetting("-linkgraph") };
    }

    @Override
    protected Class<? extends Mapper> getMapperClass() {
        return GetReady_Mapper.class;
    }

    @Override
    protected String getOutputPath(FileSystem fs) throws IOException {
        return this.getUserSetting("-output") + "/Round_0";
    }

    @Override
    protected Class<? extends Reducer> getReducerClass() {
        return null;
    }

    @Override
    protected int getReducerNumber() {
        return 0;
    }

    @Override
    protected void setUp(JobConfig conf, FileSystem fs) throws IOException {
        conf.set("output", this.getUserSetting("-output"));
        conf.set("round", 0);

        conf.set("Seeds", this.getUserSetting("-seeds"));

        BufferedReader br = new BufferedReader(new InputStreamReader(fs
                .open(new Path(this.getUserSetting("-seeds") + ".times"))));
        conf.set("SeedsTimes", br.readLine().trim());
        br.close();
    }

    @Override
    protected void tearDown(JobConfig conf, FileSystem fs, RunningJob job)
            throws IOException {
        long websize = this.getCounter(job, "WebSize", "WebSize");

        BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(fs
                .create(new Path(this.getUserSetting("-output") + "/WebSize"),
                        true)));
        bw.write(String.valueOf(websize));
        bw.flush();
        bw.close();

        fs.delete(new Path(this.getUserSetting("-output") + "/Z_0"), true);
        fs.delete(new Path(this.getUserSetting("-seeds") + ".times"), true);
    }

}
